package cn.com.sparkle.paxos.net.netlayer.raptor;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.checksum.ChecksumUtil;
import cn.com.sparkle.paxos.checksum.ChecksumUtil.UnsupportedChecksumAlgorithm;
import cn.com.sparkle.paxos.net.frame.FrameBody;
import cn.com.sparkle.paxos.net.frame.FrameHead;
import cn.com.sparkle.raptor.core.buff.BuffPool;
import cn.com.sparkle.raptor.core.buff.BuffPool.PoolEmptyException;
import cn.com.sparkle.raptor.core.buff.IoBuffer;
import cn.com.sparkle.raptor.core.io.BufferPoolOutputStream;
import cn.com.sparkle.raptor.core.io.IoBufferArrayInputStream;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.Protocol;

public class CheckSumPackageProtocol implements Protocol {
	private final static Logger logger = Logger.getLogger(CheckSumPackageProtocol.class);

	private final static byte[] nullBytes = new byte[0];

	private static class ObjectProtocolCacheBean {
		private volatile FrameHead head = null;
		private volatile int recieveSize = 0;

		private LinkedList<IoBuffer> buff = new LinkedList<IoBuffer>();
	}

	@Override
	public void init(ProtocolHandlerIoSession mySession) {
		mySession.protocolAttachment = new ObjectProtocolCacheBean();
	}

	@Override
	public Object decode(ProtocolHandlerIoSession mySession, IoBuffer buff) throws IOException {
		ObjectProtocolCacheBean bean = (ObjectProtocolCacheBean) mySession.protocolAttachment;
		if (bean.buff.size() == 0 || bean.buff.getLast() != buff) {
			bean.buff.addLast(buff);
			bean.recieveSize += buff.getByteBuffer().remaining();
		}
		IoBufferArrayInputStream inputStream = null;
		try {
			if (bean.head == null) {
				if (bean.recieveSize >= 1) {
					//unpack head
					int pos = bean.buff.get(0).getByteBuffer().position();
					byte first = bean.buff.get(0).getByteBuffer().get(pos);
					int checksumLength = FrameHead.calcChecksumLength(first);
					if (bean.recieveSize >= checksumLength + 4) {
						byte[] head = new byte[4];
						byte[] checksum;
						inputStream = new IoBufferArrayInputStream(bean.buff.toArray(new IoBuffer[bean.buff.size()]));
						inputStream.read(head);
						if (checksumLength == 0) {
							checksum = nullBytes;
						} else {
							checksum = new byte[checksumLength];
							inputStream.read(checksum);
						}
						bean.head = new FrameHead(head, checksum);
						// remove and close IoBuffer that has been unuseful.
						while (bean.buff.size() > 0 && !bean.buff.getFirst().getByteBuffer().hasRemaining()) {
							bean.buff.removeFirst();
						}
						bean.recieveSize -= checksumLength + 4;

						if (!bean.head.isValid()) {
							logger.warn("tcp checksum invalid, close connection!");
							mySession.closeSession();
							return null;
						}
					}
				}
			}

			if (bean.head != null) {
				//unpack body
				if (bean.recieveSize >= bean.head.getBodySerializeSize()) {
					if (inputStream == null) {
						inputStream = new IoBufferArrayInputStream(bean.buff.toArray(new IoBuffer[bean.buff.size()]));
					}
					byte[] body = new byte[bean.head.getBodySize()];
					inputStream.read(body);
					byte[] checksum;
					if (bean.head.getChecksumType() == ChecksumUtil.NO_CHECKSUM) {
						checksum = nullBytes;
					} else {
						checksum = new byte[bean.head.getBodyChecksumLength()];
						try {
							inputStream.read(checksum);
						} catch (RuntimeException e) {
							throw e;
						}
					}
					// remove and close IoBuffer that has been unuseful.
					while (bean.buff.size() > 0 && !bean.buff.getFirst().getByteBuffer().hasRemaining()) {
						bean.buff.removeFirst();
					}

					bean.recieveSize -= body.length + checksum.length;
					FrameBody frameBody = new FrameBody(body, checksum, bean.head.getChecksumType());
					bean.head = null;//clear last time head 
					return frameBody;
				}

			}
			return null;
		} finally {
			if (inputStream != null) {
				inputStream.close();
			}
		}

	}

	@Override
	public IoBuffer[] encode(BuffPool buffpool, Object message) throws IOException, PoolEmptyException {
		return encode(buffpool, message, null);
	}

	@Override
	public IoBuffer[] encode(BuffPool buffpool, Object message, IoBuffer lastWaitSendBuff) throws IOException, PoolEmptyException {

		FrameBody body = (FrameBody) message;
		FrameHead head;
		try {
			head = new FrameHead(body.getChecksumType(), body.getBodySize());
		} catch (UnsupportedChecksumAlgorithm e) {
			throw new IOException(e);
		}
		BufferPoolOutputStream bufferPoolOutputStream = new BufferPoolOutputStream(buffpool, 0, lastWaitSendBuff);
		bufferPoolOutputStream.write(head.getHead());
		bufferPoolOutputStream.write(head.getChecksum());
		bufferPoolOutputStream.write(body.getBody());
		bufferPoolOutputStream.write(body.getChecksum());
		List<IoBuffer> list = bufferPoolOutputStream.getBuffArray();
		if (lastWaitSendBuff != null) {
			list.remove(0);
		}
		bufferPoolOutputStream.close();
		return list.toArray(new IoBuffer[list.size()]);
	}

}
